package com.wisecoach.gatewayplus.event;

import com.google.protobuf.InvalidProtocolBufferException;
import com.wisecoach.util.Assert;
import io.grpc.CallOptions;
import org.hyperledger.fabric.client.CloseableIterator;
import org.hyperledger.fabric.client.GatewayException;
import org.hyperledger.fabric.client.Network;
import org.hyperledger.fabric.protos.common.BlockchainInfo;
import org.hyperledger.fabric.protos.peer.FilteredBlock;

import java.util.concurrent.ExecutorService;
import java.util.function.UnaryOperator;

/**
 * {@code @author:} wisecoach
 * {@code @date:} 2023/6/15 上午2:36
 * {@code @version:} 1.0.0
 */
public class DefaultFilteredBlockSource implements FilteredBlockSource {

    private final String channelName;
    private final Network network;
    private final ExecutorService executorService;
    private FilteredBlockEventPublisher publisher;
    private volatile Long cache;

    public DefaultFilteredBlockSource(String channelName, Network network, ExecutorService executorService) {
        this.channelName = channelName;
        this.network = network;
        this.executorService = executorService;
    }

    @Override
    public String getChannel() {
        return channelName;
    }

    @Override
    public void setFilteredBlockEventPublisher(FilteredBlockEventPublisher publisher) {
        this.publisher = publisher;
    }

    @Override
    public long getLatest() {
        if (cache == null) {
            try {
                byte[] bytes = network.getContract("qscc").evaluateTransaction("GetChainInfo", channelName);
                BlockchainInfo blockchainInfo = BlockchainInfo.parseFrom(bytes);
                cache = blockchainInfo.getHeight();
            } catch (GatewayException | InvalidProtocolBufferException e) {
                throw new RuntimeException("无法取得当前最新高度", e);
            }
        }
        return cache;
    }

    @Override
    public void start(UnaryOperator<CallOptions> options) {
        start(0, options);
    }

    @Override
    public void start(long startNumber, UnaryOperator<CallOptions> options) {
        Assert.notNull(publisher, "执行前请先设置要将事件发送给哪个publisher");
        cache = null;
        executorService.execute(() -> {
            CloseableIterator<FilteredBlock> events = network.newFilteredBlockEventsRequest().startBlock(startNumber).build().getEvents();
            events.forEachRemaining(event -> {
                FilteredBlockEvent filteredBlockEvent = new FilteredBlockEvent(event);
                publisher.publish(filteredBlockEvent);
            });
        });
    }
}
